package edu.utexas.mapreduce;

import java.io.BufferedWriter;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.util.Map;
import java.util.TreeMap;


public class MapOutputWriter implements OutputWriter {
	
	private final String outputDir;
	private final int sequence;
	private Map<Integer, PrintWriter> writerMap = new TreeMap<Integer, PrintWriter>();
	
	MapOutputWriter(String outputDir, int sequence) {
		this.outputDir = outputDir;
		this.sequence = sequence;
	}
	
	/**
	 * 
	 */
	public synchronized void write(String key, String value) {
		try {
			//
			// Create one BufferedWriter per reduce keyID.  The
			// buffered writer will do the buffering and flushing
			// to disk.  This is a LOT faster than keeping a map
			// of keys and then flushing each key individually
			//
			Integer keyID = new Integer(ReduceTask.getKeyID(key));

			if (!writerMap.containsKey(keyID)) {
				String filepath = outputDir + getFileName(keyID);
				OutputStreamWriter os = new OutputStreamWriter(new FileOutputStream(filepath, false), XML.ENCODING);
				writerMap.put(keyID, new PrintWriter(new BufferedWriter(os)));
				
				// Write XML header and DTD
				PrintWriter out = writerMap.get(keyID);
				out.println("<?xml version=\"1.0\" encoding=\"" + XML.ENCODING + "\"?>");
				out.println("<!DOCTYPE map_output [");
				out.println("  <!ELEMENT map_output (map)*>");
				out.println("  <!ELEMENT map (key, value)>");
				out.println("  <!ELEMENT key (#PCDATA)>");
				out.println("  <!ELEMENT value (#PCDATA)>");
				out.println("]>");
				out.println("<map_output>");
			}

			PrintWriter out = writerMap.get(keyID);
			out.write("<map>");
			out.write("<key><![CDATA[" + key + "]]></key>");
			out.write("<value><![CDATA[" + value + "]]></value>");
			out.write("</map>");
			out.println();
		}
		catch(IOException e) {
			System.out.println("Mapper output file error:");
			e.printStackTrace();
		}
	}
	
	void close() {
		for (PrintWriter out: writerMap.values()) {
			out.println("</map_output>");
			out.flush();
			out.close();
		}
	}
	
	String getFileName(Integer keyID) {
		return "map_" + String.format("%04d", sequence) + "_"+ String.format("%04d", keyID);
	}
	
	Map<Integer, String> getOutputFiles() {
		Map<Integer, String> filemap = new TreeMap<Integer, String>();
		for(Integer keyID: writerMap.keySet()) {
			filemap.put(keyID, getFileName(keyID));
		}
		return filemap;
	}
	
}
